Chip 2007 January, February, March & April/Chip-Cover-CD-2007-02.iso/Pakiet bezpieczenstwa/mini Pentoo LiveCD 2006.1/mpentoo-2006.1.iso/livecd.squashfs/usr/lib/python2.4/test/test_queue.pyo

Python Compiled Bytecode | 2005-10-18 | 6KB | 247 lines

# Source Generated with Decompyle++
# File: in.pyo (Python 2.4)

import Queue
import sys
import threading
import time
from test.test_support import verify, TestFailed, verbose
QUEUE_SIZE = 5

class _TriggerThread(threading.Thread):

    def __init__(self, fn, args):
        self.fn = fn
        self.args = args
        self.startedEvent = threading.Event()
        threading.Thread.__init__(self)


    def run(self):
        time.sleep(0.10000000000000001)
        self.startedEvent.set()
        self.fn(*self.args)



def _doBlockingTest(block_func, block_args, trigger_func, trigger_args):
    t = _TriggerThread(trigger_func, trigger_args)
    t.start()
    result = block_func(*block_args)
    if not t.startedEvent.isSet():
        raise TestFailed("blocking function '%r' appeared not to block" % block_func)

    t.join(10)
    if t.isAlive():
        raise TestFailed("trigger function '%r' appeared to not return" % trigger_func)

    return result


def _doExceptionalBlockingTest(block_func, block_args, trigger_func, trigger_args, expected_exception_class):
    t = _TriggerThread(trigger_func, trigger_args)
    t.start()

    try:
        block_func(*block_args)
    except expected_exception_class:
        raise
    else:
        raise TestFailed('expected exception of kind %r' % expected_exception_class)
    finally:
        t.join(10)
        if t.isAlive():
            raise TestFailed("trigger function '%r' appeared to not return" % trigger_func)

        if not t.startedEvent.isSet():
            raise TestFailed('trigger thread ended but event never set')




class FailingQueueException(Exception):
    pass


class FailingQueue(Queue.Queue):

    def __init__(self, *args):
        self.fail_next_put = False
        self.fail_next_get = False
        Queue.Queue.__init__(self, *args)


    def _put(self, item):
        if self.fail_next_put:
            self.fail_next_put = False
            raise FailingQueueException, 'You Lose'

        return Queue.Queue._put(self, item)


    def _get(self):
        if self.fail_next_get:
            self.fail_next_get = False
            raise FailingQueueException, 'You Lose'

        return Queue.Queue._get(self)



def FailingQueueTest(q):
    if not q.empty():
        raise RuntimeError, 'Call this function with an empty queue'

    for i in range(QUEUE_SIZE - 1):
        q.put(i)

    q.fail_next_put = True

    try:
        q.put('oops', block = 0)
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    q.fail_next_put = True

    try:
        q.put('oops', timeout = 0.10000000000000001)
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    q.put('last')
    verify(q.full(), 'Queue should be full')
    q.fail_next_put = True

    try:
        _doBlockingTest(q.put, ('full',), q.get, ())
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    q.put('last')
    q.fail_next_put = True

    try:
        _doExceptionalBlockingTest(q.put, ('full', True, 10), q.get, (), FailingQueueException)
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    q.put('last')
    verify(q.full(), 'Queue should be full')
    q.get()
    verify(not q.full(), 'Queue should not be full')
    q.put('last')
    verify(q.full(), 'Queue should be full')
    _doBlockingTest(q.put, ('full',), q.get, ())
    for i in range(QUEUE_SIZE):
        q.get()

    verify(q.empty(), 'Queue should be empty')
    q.put('first')
    q.fail_next_get = True

    try:
        q.get()
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    verify(not q.empty(), 'Queue should not be empty')
    q.fail_next_get = True

    try:
        q.get(timeout = 0.10000000000000001)
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    verify(not q.empty(), 'Queue should not be empty')
    q.get()
    verify(q.empty(), 'Queue should be empty')
    q.fail_next_get = True

    try:
        _doExceptionalBlockingTest(q.get, (), q.put, ('empty',), FailingQueueException)
        raise TestFailed("The queue didn't fail when it should have")
    except FailingQueueException:
        pass

    verify(not q.empty(), 'Queue should not be empty')
    q.get()
    verify(q.empty(), 'Queue should be empty')


def SimpleQueueTest(q):
    if not q.empty():
        raise RuntimeError, 'Call this function with an empty queue'

    q.put(111)
    q.put(222)
    if q.get() == 111:
        pass
    verify(q.get() == 222, "Didn't seem to queue the correct data!")
    for i in range(QUEUE_SIZE - 1):
        q.put(i)
        verify(not q.empty(), 'Queue should not be empty')

    verify(not q.full(), 'Queue should not be full')
    q.put('last')
    verify(q.full(), 'Queue should be full')

    try:
        q.put('full', block = 0)
        raise TestFailed("Didn't appear to block with a full queue")
    except Queue.Full:
        pass


    try:
        q.put('full', timeout = 0.01)
        raise TestFailed("Didn't appear to time-out with a full queue")
    except Queue.Full:
        pass

    _doBlockingTest(q.put, ('full',), q.get, ())
    _doBlockingTest(q.put, ('full', True, 10), q.get, ())
    for i in range(QUEUE_SIZE):
        q.get()

    verify(q.empty(), 'Queue should be empty')

    try:
        q.get(block = 0)
        raise TestFailed("Didn't appear to block with an empty queue")
    except Queue.Empty:
        pass


    try:
        q.get(timeout = 0.01)
        raise TestFailed("Didn't appear to time-out with an empty queue")
    except Queue.Empty:
        pass

    _doBlockingTest(q.get, (), q.put, ('empty',))
    _doBlockingTest(q.get, (True, 10), q.put, ('empty',))


def test():
    q = Queue.Queue(QUEUE_SIZE)
    SimpleQueueTest(q)
    SimpleQueueTest(q)
    if verbose:
        print 'Simple Queue tests seemed to work'

    q = FailingQueue(QUEUE_SIZE)
    FailingQueueTest(q)
    FailingQueueTest(q)
    if verbose:
        print 'Failing Queue tests seemed to work'


test()